package com.winchain.provider.expiration.rx;

import com.winchain.chainbill.basic.cus.po.CusGrp;
import com.winchain.chainbill.basic.cus.service.CusGrpService;
import com.winchain.provider.expiration.service.BillExpThreadService;
import com.winchain.provider.expiration.vo.ExpResult;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Future;
import org.apache.commons.logging.Log;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import rx.Observable;

@Component
public class BillExpTaskSubsciber extends TransactionNewSubscriber<BillExpirationEvent> {
	@Autowired
	private CusGrpService cusGrpService;
	@Autowired
	private BillExpThreadService billExpThreadService;
	@Autowired
	private BillExpSettlSubscriber billExpSettlSubscriber;
	@Autowired
	private BillExpDebitSubscriber billExpDebitSubscriber;
	@Autowired
	private BillExpCreditSubscriber billExpCreditSubscriber;
	@Autowired
	private BillExpDelayInsSubscriber billExpDelayInsSubscriber;
	@Autowired
	private BillExpAssetSubscriber billExpAssetSubscriber;

	public void execute(BillExpirationEvent t) {
		List<CusGrp> grps = this.cusGrpService.selectAll();

		CountDownLatch counter = new CountDownLatch(grps.size());
		List<Future<ExpResult>> futures = new ArrayList();
		for (CusGrp grp : grps) {
			BillExpirationEvent event = new BillExpirationEvent();
			event.setDate(t.getDate());
			event.setPm(t.isPm());
			event.setGrpId(grp.getGrpId());
			Future<ExpResult> future = (Future<ExpResult>) this.billExpThreadService
					.submit(new ExpTask(counter, event));
			futures.add(future);
		}
		t.setValue("counter", counter);
		t.setValue("futures", futures);
	}

	private void process(BillExpirationEvent event) {
		Observable<BillExpirationEvent> observers = Observable.just(event);
		observers.subscribe(this.billExpSettlSubscriber);
		observers.subscribe(this.billExpDelayInsSubscriber);
		observers.subscribe(this.billExpDebitSubscriber);
		observers.subscribe(this.billExpCreditSubscriber);
		observers.subscribe(this.billExpAssetSubscriber);
	}

	class ExpTask implements Callable<ExpResult> {
		private CountDownLatch counter;
		private BillExpirationEvent event;

		public ExpTask(CountDownLatch counter, BillExpirationEvent event) {
			this.counter = counter;
			this.event = event;
		}

		public ExpResult call() throws Exception {
			ExpResult result = new ExpResult();
			try {
				BillExpTaskSubsciber.this.process(this.event);

				Object debits = this.event.getValue("debits");
				if (debits != null) {
					result.setDebits((List) debits);
				}
				return result;
			} catch (Exception e) {
				ExpResult localExpResult1;
				BillExpTaskSubsciber.this.logger.error(e);
				return result;
			} finally {
				this.counter.countDown();
			}
		}
	}
}
